package com.wsjj.gmall.util;

import com.wsjj.gmall.constant.Constant;

//  flinkSQL 工具类
public class SQLUtil {


    /**
     * flinkSQL连接kafka
     * @param topic 连接的topic
     * @param group 连接的group
     * @return
     */
    public static String getKafkaDDL(String topic,String group){
        return "WITH (\n" +
                "  'connector' = 'kafka',\n" +
                        "  'topic' = '"+topic+"',\n" +
                        "  'properties.bootstrap.servers' = '"+ Constant.KAFKA_BROKERS +"',\n" +
                        "  'properties.group.id' = '"+group+"',\n" +
                        "  'scan.startup.mode' = 'earliest-offset',\n" +
                        "  'format' = 'json'\n" +
                        ")";
    }

    /**
     * flinkSQL连接HBase
     * @param namespace  表空间（相当于mysql的数据库）
     * @param table   表
     * @return
     */
    public static String getHbaseDDL(String namespace,String table){
        return "WITH (\n" +
                " 'connector' = 'hbase-2.2',\n" +
                        " 'table-name' = '"+ namespace+":"+table+"',\n" +
                        " 'zookeeper.quorum' = '"+Constant.ZOOKEEPER_QUORUM+"',\n" +
                        " 'lookup.async' = 'true',\n" +
                        " 'lookup.cache' = 'PARTIAL',\n" +
                        " 'lookup.partial-cache.max-rows' = '500',\n" +
                        " 'lookup.partial-cache.expire-after-access' = '1 hour ',\n" +
                        " 'lookup.partial-cache.expire-after-write' = '1 hour '\n" +
                        ")";

    }


    /**
     * flinkSQL 使用update的方式连接kafka
     * @param topic topic
     * @return
     */
    public static String getUpdateKafkaDDL(String topic){
        return "WITH (\n" +
                "  'connector' = 'upsert-kafka',\n" +
                        "  'topic' = '"+topic+"',\n" +
                        "  'properties.bootstrap.servers' = '"+Constant.KAFKA_BROKERS+"',\n" +
                        "  'key.format' = 'json',\n" +
                        "  'value.format' = 'json'\n" +
                        ")";

    }



}
